package rxgo

import (
	"context"
	"encoding/json"
	"fmt"
	"reflect"
)

type filteringOperator struct {
	opFunc func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool)
}

func (fiop filteringOperator) op(ctx context.Context, o *Observable) {
	in := o.pred.outflow
	out := o.outflow

	go func() {
		end := false
		for x := range in {

			if end {
				// continue
				break
			}
			// can not pass a interface as parameter (pointer) to gorountion for it may change its value outside!
			xv := reflect.ValueOf(x)
			// send an error to stream if the flip not accept error
			if e, ok := x.(error); ok && !o.flipAcceptError {
				o.sendToFlow(ctx, e, out)
				continue
			}
			// fmt.Println("op", x)
			if fiop.opFunc(ctx, o, xv, out) {
				end = true
			}
		}
		// this should be the end
		// and we send out buffers to outflow
		if o.flip != nil {
			buffer := (reflect.ValueOf(o.flip))
			if buffer.Kind() != reflect.Slice {
				panic("flip is not buffer")
			}
			for i := 0; i < buffer.Len(); i++ {
				o.sendToFlow(ctx, buffer.Index(i).Interface(), out)
			}
		}

		o.closeFlow(out)
	}()

}

// Take filter out first <num> items from parent
func (parent *Observable) Take(num int) (o *Observable) {
	o = parent.newFilterObservable("take.n")
	o.flipAcceptError = true
	o.flipSupCtx = true

	// 初始化计数
	count := 0
	// 具备记忆功能的operator不可避免的两个方式：
	// operator 自带记忆变量
	// 或者通过外部变量逃逸
	o.operator = filteringOperator{
		opFunc: func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool) {

			count++

			if count <= num {
				o.sendToFlow(ctx, item.Interface(), out)
				return false
			}
			return true
		},
	}

	return o
}

// TakeLast filter out last <num> items from parent
func (parent *Observable) TakeLast(num int) (o *Observable) {
	o = parent.newFilterObservable("takeLast.n")
	o.flipAcceptError = true
	o.flipSupCtx = true

	// 初始化计数
	count := 0
	var lasts []interface{}
	// 具备记忆功能的operator不可避免的两个方式：
	// operator 自带记忆变量
	// 或者通过外部变量逃逸
	o.operator = filteringOperator{
		opFunc: func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool) {

			count++
			// 维护一个长度为num的buffer
			if count <= num {
				lasts = append(lasts, item.Interface())
			} else {
				lasts = lasts[1:]
				lasts = append(lasts, item.Interface())
			}
			// fmt.Println("lasts", lasts)
			o.flip = lasts
			return false
		},
	}

	return o
}

// TakeFirst skip out first item that satisfy f(item) from parent
func (parent *Observable) TakeFirst(f interface{}) (o *Observable) {

	// check func
	fv := reflect.ValueOf(f)
	inType := []reflect.Type{typeAny}
	outType := []reflect.Type{typeBool}
	b, _ := checkFuncUpcast(fv, inType, outType, true)
	if !b {
		panic(ErrFuncFlip)
	}
	// o.flip = fv.Interface()
	// o.flipSupCtx = ctxSup

	o = parent.newFilterObservable("takeFirst.f")
	o.flipAcceptError = true
	o.flipSupCtx = true

	o.operator = filteringOperator{
		opFunc: func(ctx context.Context, o *Observable, x reflect.Value, out chan interface{}) (end bool) {

			var params = []reflect.Value{x}
			rs, skip, stop, e := userFuncCall(fv, params)
			var satisfy = rs[0].Interface().(bool)

			if stop {
				end = true
				return
			}
			if skip {
				return
			}
			if e != nil {
				end = o.sendToFlow(ctx, e, out)
				if end {
					return
				}
				return
			}

			if satisfy {
				o.sendToFlow(ctx, x.Interface(), out)
				return true
			}

			return false
		},
	}
	return o
}

// Skip skip out first <num> items from parent
func (parent *Observable) Skip(num int) (o *Observable) {
	o = parent.newFilterObservable("skip.n")
	o.flipAcceptError = true
	o.flipSupCtx = true
	count := 0
	o.operator = filteringOperator{
		opFunc: func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool) {

			count++

			if count > num {
				o.sendToFlow(ctx, item.Interface(), out)
			}
			return false
		},
	}

	return o
}

// Distinct only filter out distinct elements
func (parent *Observable) Distinct() (o *Observable) {
	o = parent.newFilterObservable("distinct")
	o.flipAcceptError = true
	o.flipSupCtx = true
	// 亮点设计：
	// struct{}作为value可以避免内存过大
	m := map[string]struct{}{}

	o.operator = filteringOperator{
		opFunc: func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool) {

			// 简单序列化版本
			seril := fmt.Sprintf("%v", item)
			// 改进，最好是提供json的序列化
			bytes, err := json.Marshal(item)
			if err != nil {
				seril = string(bytes)
			}

			if _, ok := m[seril]; !ok {
				m[seril] = struct{}{}
				o.sendToFlow(ctx, item.Interface(), out)
			}
			return false
		},
	}

	return o
}

func (parent *Observable) newFilterObservable(name string) (o *Observable) {
	//new Observable
	o = newObservable()
	o.Name = name

	//chain Observables
	parent.next = o
	o.pred = parent
	o.root = parent.root

	//set options
	o.bufLen = BufferLen
	return o
}
